package cn.com.sparkle.paxos.deamon;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.NodesCollection;
import cn.com.sparkle.paxos.event.CatchUpEvent;
import cn.com.sparkle.paxos.event.ConfigureEvent;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionId;
//import cn.com.sparkle.paxos.model.ElectionId;
//import cn.com.sparkle.paxos.model.Id;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.paxosinstance.ElectionPaxosInstance;
import cn.com.sparkle.paxos.util.IdComparator;

public class CheckMasterSenatorStateDeamon implements Runnable {
	private final static Logger logger = Logger
			.getLogger(CheckMasterSenatorStateDeamon.class);

	private ClusterState cState;
	// assume this system's state is failure when initiate this system.
	private boolean checkable = false;
	// private ClientNetHandler handler;
	private ReentrantLock lock = new ReentrantLock();

	private String selfAddress;

	public CheckMasterSenatorStateDeamon(final ClusterState cState,
			Configuration conf, EventsManager eventsManager) {
		this.cState = cState;
		this.selfAddress = conf.getSelfAddress();
		eventsManager.registerEvent(new CatchUpEvent() {
			@Override
			public void recoveryFromFail() {
				try {
					lock.lock();
					checkable = true;
				} finally {
					lock.unlock();
				}
			}

			@Override
			public void catchUpFail() {
				try {
					lock.lock();
					checkable = false;
				} finally {
					lock.unlock();
				}

			}
		});

	}

	@Override
	public void run() {
		while (true) {
			try {
				Thread.sleep(500);
			} catch (InterruptedException e) {
			}
			try {
				lock.lock();
				// memory barrier
				if (!checkable) {
					continue;
				}
				NodesCollection senators = cState.getSenators();
				int quorum = senators.getNodeMembers().size() / 2 + 1;
				Set<NetNode> activeSenator = senators.getActiveNodes();
				ElectionId maxElectionid = cState.getLastElectionId();
				int maxElectionIdLostCount = 0;
				for (NetNode nnode : activeSenator) {
					NodeState ns = nnode.getNodeState();
					if (ns.getLastElectionId().getVersion() != maxElectionid
							.getVersion()
							|| (IdComparator.getInstance().compare(
									ns.getLastElectionId().getId(),
									maxElectionid.getId()) == 0 && !ns
									.isMasterConnected())) {
						++maxElectionIdLostCount;
					} else if (IdComparator.getInstance().compare(
							ns.getLastElectionId().getId(),
							maxElectionid.getId()) == 1) {
						maxElectionid = ns.getLastElectionId();
						maxElectionIdLostCount = 0;
					}
				}
				if (!selfAddress.equals(maxElectionid.getId().getAddress())) {
					// active study newest master
					cState.changeLastElectionId(maxElectionid);
				}
				// active lost the position of master
				if (activeSenator.size() < quorum) {
					cState.lostAuthorizationOfMaster();
				}
				// if and only if the number that itself can connect more than
				// quorum , it can elect itself as master.
				// And when the number that itself can connect more than quorum
				// , it
				// must can get information from others about itself has been or
				// not
				// a master.
				// If it has been a master and it lost its position actively, it
				// must start a election that elect itself as a master again to
				// remind others that the master has been lost.
				if (maxElectionIdLostCount >= quorum
						|| (cState.getLastElectionId().getId().getIncreaseId() == -1 && selfAddress
								.equals(maxElectionid.getId().getAddress()))) {
					logger.debug("master is lost!!!");
					startSelection(activeSenator, quorum);
				}
			} finally {
				lock.unlock();
			}
		}
	}

	private void startSelection(Set<NetNode> activeSenator, int quorum) {
		// fix address to selfaddress
		cState.getSelfState().getElectionVoteIdBySelf().getIdBuilder()
				.setAddress(selfAddress);
		logger.debug("start election, instanceId:"
				+ cState.getSelfState().getElectionVoteIdBySelf().getId()
						.getIncreaseId());
		ElectionPaxosInstance paxosInstance = new ElectionPaxosInstance(
				activeSenator, quorum, cState.getSelfState()
						.getElectionVoteIdBySelf().build(), cState, selfAddress);
		Future<Boolean> f = paxosInstance.activate();
		boolean isSuccess = false;
		try {
			isSuccess = f.get();
		} catch (InterruptedException e1) {
			logger.error("fatal error", e1);
		} catch (ExecutionException e1) {
			logger.error("fatal error", e1);
		}
		logger.debug("election voteId.increaseId:"
				+ cState.getSelfState().getElectionVoteIdBySelf().getId()
						.getIncreaseId() + "  election isSuccessful:"
				+ isSuccess);
		if (!isSuccess) {
			try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
			}
		}

	}
}
